package com.atguigu.gmall.realtime.test;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @author Felix
 * @date 2023/11/11
 * 该案例演示了Watermark的传递
 */
public class Flink04_Watermark {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(2);
        //TODO 2.检查点相关的设置(略)

        //TODO 3.从指定的网络端口读取员工数据
        env
            .socketTextStream("hadoop102", 8888)
            .map(
                lineStr -> {
                    String[] fieldArr = lineStr.split(",");
                    return new Emp(Integer.valueOf(fieldArr[0]), fieldArr[1], Integer.valueOf(fieldArr[2]), Long.valueOf(fieldArr[3]));
                }
            ).assignTimestampsAndWatermarks(
                // WatermarkStrategy.<Emp>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                WatermarkStrategy
                    .<Emp>forMonotonousTimestamps()
                    .withTimestampAssigner(
                        new SerializableTimestampAssigner<Emp>() {
                            @Override
                            public long extractTimestamp(Emp emp, long recordTimestamp) {
                                return emp.getTs();
                            }
                        }
                    )
            ).keyBy(Emp::getEmpno)
            .window(TumblingEventTimeWindows.of(Time.milliseconds(10)))
            .process(
                new ProcessWindowFunction<Emp, Emp, Integer, TimeWindow>() {
                    @Override
                    public void process(Integer integer, Context context, Iterable<Emp> elements, Collector<Emp> out) throws Exception {
                        for (Emp emp : elements) {
                            out.collect(emp);
                        }
                    }
                }
            ).print();
        env.execute();
    }
}
